package org.sky.cloud.common.task;


import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;

public class DelayTaskExecuteEngine<T extends AbstractDelayTask<?>> implements TaskExecuteEngine<T> {

    protected ScheduledExecutorService processTaskExecutor;

    protected final ConcurrentHashMap<String, T> tasks;

    private final ReentrantLock lock = new ReentrantLock();

    private DelayTaskProcessor<T> processor;

    public DelayTaskExecuteEngine() {
        this.tasks = new ConcurrentHashMap<>(32);
        processTaskExecutor = Executors.newSingleThreadScheduledExecutor();
        processTaskExecutor.scheduleWithFixedDelay(new ProcessRunnable(), 100L, 100L, TimeUnit.MILLISECONDS);
    }

    @Override
    public int size() {
        lock.lock();
        try {
            return tasks.size();
        }finally {
            lock.unlock();
        }
    }


    @Override
    public void registerProcessor(DelayTaskProcessor<T> processor){
        this.processor = processor;
    }

    @Override
    public boolean isEmpty() {
        return tasks.isEmpty();
    }

    @Override
    public void addTask(String key, T task) {
        lock.lock();
        try {
            tasks.put(key, task);
        } finally {
            lock.unlock();
        }
    }




    @Override
    public T removeTask(String key) {
        lock.lock();
        try{
            T task = tasks.get(key);
            if(task != null && task.shouldProcess()){
                return tasks.remove(key);
            }else{
                return null;
            }

        } finally {
          lock.unlock();
        }
    }

    private class ProcessRunnable implements Runnable{

        @Override
        public void run() {
            processTask();
        }

    }


    private void processTask() {
        for(String key : tasks.keySet()){
            T delayTask = removeTask(key);
            if(delayTask != null){
                processor.process(delayTask);
            }
        }
    }

}
